package com.xlhj.boot.controller;

import com.aliyun.odps.Column;
import com.aliyun.odps.OdpsType;
import com.aliyun.odps.TableSchema;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.RecordReader;
import com.aliyun.odps.data.ResultSet;
import com.aliyun.odps.tunnel.InstanceTunnel;
import com.aliyun.odps.tunnel.io.TunnelRecordReader;
import com.xlhj.boot.common.ResultData;
import com.xlhj.boot.service.MaxComputeService;
import com.xlhj.boot.service.OssService;
import com.xlhj.boot.util.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.ResultSet;
import com.xlhj.boot.common.ObjectResult;
import com.xlhj.boot.common.ResultData;
import com.xlhj.boot.service.MaxComputeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * @author: liucaijing
 * @description: MaxComputeController
 * @date: 2021/6/18 9:50
import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author: caijliu
 * @description: MaxComputeController
 * @data: 23/06/2021 9:44
 */
@RestController
public class MaxComputeController {

    private Logger logger = LoggerFactory.getLogger(MaxComputeController.class);

    @Autowired
    private MaxComputeService maxComputeService;
    @Autowired
    private OssService ossService;

    /**
     * 根据表名查询数据
     * @param tableName
     * @return
     */
    @GetMapping("/getDataInfo/{tableName}")
    public ResultData getDataInfo(@PathVariable String tableName) {
        logger.info("This is request param: ", tableName);
        ResultData resultData = new ResultData();
        try {
            ResultSet records = maxComputeService.getDataInfo(tableName);
            List<Map<String, Object>> dataList = new ArrayList<>();
            if (null != records && records.hasNext()) {
                Iterator<Record> iterator = records.iterator();
                //遍历返回结果
                while (iterator.hasNext()) {
                    Map<String, Object> dataMap = new HashMap<>();
                    Record record = iterator.next();
                    Column[] columns = record.getColumns();
                    for (Column column : columns) {
                        dataMap = getColumnValue(column, record, dataMap);
                    }
                    dataList.add(dataMap);
                }
            } else {
                //如果没有查询到数据，返回表列信息
                List<Column> columns = maxComputeService.getTableColumn(tableName);
                if (null != columns && columns.size() > 0) {
                    Map<String, Object> dataMap = new HashMap<>();
                    for (Column column : columns) {
                        String name = column.getName();
                        dataMap.put(name, "");
                    }
                    dataList.add(dataMap);
                }
            }
            logger.info("查询maxcompute数据为：", Arrays.toString(dataList.toArray()));
            resultData = ResultData.success().data(dataList);
        } catch (Exception e) {
            logger.error("查询maxcompute数据异常：", e.getMessage());
            resultData = ResultData.fail().message(e.getMessage());
        }
        return resultData;
    }

    @GetMapping("/uploadData")
    public ResultData uploadData() throws UnsupportedEncodingException {
        ossService.appendDataToOss(null, 0);
        return ResultData.success();
    }

    @GetMapping("/getDataSize")
    public ResultData getDataSize() {
        ossService.getFileInfoForMarker();
        return ResultData.success();
    }

    @GetMapping("/getBatchDataInfoForThread/{tableName}")
    public ResultData getBatchDataInfoForThread(@PathVariable String tableName) {
        ResultData resultData = new ResultData();
        try {
            InstanceTunnel.DownloadSession downloadSession = maxComputeService.getBatchDataInfo(tableName);
            System.out.println("Session Status is ..." + downloadSession.getStatus().toString());
            long recordCount = downloadSession.getRecordCount();
            System.out.println("RecordCount is : " + recordCount);
            //ExecutorService pool = Executors.newFixedThreadPool(10);
            ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(10, 15, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(20), new ThreadPoolExecutor.CallerRunsPolicy());
            ArrayList<Callable<Long>> callables = new ArrayList<>();
            long start = 0;
            long step = recordCount / 10;
            for (int i = 0; i < 10 - 1; i++) {
                RecordReader recordReader = downloadSession.openRecordReader(step * i, step);
                callables.add(new DownloadThread(i, recordReader, downloadSession.getSchema()));
            }
            RecordReader recordReader = downloadSession.openRecordReader(step * (10 - 1), recordCount - ((10 - 1) * step));
            callables.add(new DownloadThread(10 - 1, recordReader, downloadSession.getSchema()));
            Long downloadNum = 0L;
            List<Future<Long>> recordNum = poolExecutor.invokeAll(callables);
            for (Future<Long> num : recordNum) {
                downloadNum += num.get();
            }
            System.out.println("Record Count is: " + downloadNum);
            poolExecutor.shutdown();
        } catch (Exception e ) {
            e.printStackTrace();
        }
        return resultData;
    }

    @GetMapping("/getBatchDataInfo/{tableName}")
    public ResultData getBatchDataInfo(@PathVariable String tableName) {
        ResultData resultData = new ResultData();
        try {
            List<Map<String, Object>> dataList = new ArrayList<>();
            InstanceTunnel.DownloadSession downloadSession = maxComputeService.getBatchDataInfo(tableName);
            //总条数
            long recordCount = downloadSession.getRecordCount();
            //解析downloadSession中的数据
            TunnelRecordReader recordReader = downloadSession.openRecordReader(0, recordCount);
            //先追加自动
            String content = "user_id,user_name,phone,user_addr,create_time\n";
            ossService.appendDataToOss(content, 0);
            Record record;
            //遍历查询的数据
            StringJoiner stringJoiner = new StringJoiner("");
            int count = 0;
            while (null != (record = recordReader.read())) {
                count++;
                TableSchema schema = downloadSession.getSchema();
                StringJoiner joiner = new StringJoiner(",");
                for (int i = 0; i < schema.getColumns().size(); i++) {
                    Column column = schema.getColumn(i);
                    String colValue = "";
                    if (column.getTypeInfo().getOdpsType() == OdpsType.BIGINT) {
                        Long col_value = record.getBigint(column.getName());
                        colValue = col_value == null ? null : col_value.toString();
                    }
                    if (column.getTypeInfo().getOdpsType() == OdpsType.STRING) {
                        String col_value = record.getString(column.getName());
                        colValue = col_value == null ? null : col_value;
                    }
                    if (column.getTypeInfo().getOdpsType() == OdpsType.DATETIME) {
                        Date col_value = record.getDatetime(column.getName());
                        colValue = col_value == null ? null : DateUtils.getDateToStr(col_value);
                    }
                    joiner.add(colValue);
                }
                stringJoiner.add(joiner.toString() + "\n");
                if (count%4 == 0 || count ==recordCount) {
                    System.out.println(stringJoiner.toString());
                    ossService.appendDataToOss(stringJoiner.toString(), 1);
                    stringJoiner = new StringJoiner("");
                }
            }
            resultData = ResultData.success().data(dataList);
        } catch (Exception e) {
            logger.info("查询数据异常：", e);
    @Autowired
    private MaxComputeService maxComputeService;

    @GetMapping("/runSqlTask")
    public void runSqlTask() {
        maxComputeService.runSqlTask();
    }

    @GetMapping("/dowloadData")
    public void dowloadData() {
        maxComputeService.dowloadData();
    }

    @GetMapping("/uploadData")
    public void uploadData() {
        maxComputeService.uploadData();
    }

    /**
     * 查询DATA API详情数据
     * @param tableName
     * @param token
     * @return
     */
    @GetMapping("/getDataAssetDetail/{tableName}/{token}")
    public ObjectResult getDataAssetDetail(@PathVariable String tableName, @PathVariable String token) {
        ObjectResult result = new ObjectResult();
        try {
            //解析用户信息
            //todo
            //根据用户信息查询用户所拥有的权限
            //todo
            //根据用户权限查询数据
            ResultSet recordList = maxComputeService.getDataAssetDetail();
            List<Map<String, Object>> dataList = new ArrayList<>();
            if (null != recordList && recordList.hasNext()) {
                Iterator<Record> iterator = recordList.iterator();
                //遍历数据
                while (iterator.hasNext()) {
                    Map<String, Object> map = new HashMap<String, Object>();
                    Record record = iterator.next();
                    Column[] columns = record.getColumns();
                    for (Column column : columns) {
                        map = getColumnValue(column, record, map);
                    }
                    dataList.add(map);
                }
            } else {
                //如果没有数据时，获取表列信息
                List<Column> columnList = maxComputeService.getTableColumn();
                if (null != columnList && columnList.size() > 0) {
                    Map<String, Object> map = new HashMap<>();
                    for (Column column : columnList) {
                        String name = column.getName();
                        map.put(name, null);
                    }
                    dataList.add(map);
                }
            }
            //查询下载地址
            //todo
            result.success(dataList);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }

    @GetMapping("/getDowloadDataAsset/{tableName}/{token}/{startDate}/{endDate}")
    public ResultData getDowloadDataAsset(@PathVariable String tableName, @PathVariable String token, @PathVariable String startDate, @PathVariable String endDate) {
        ResultData resultData = new ResultData();
        try {
            //解析用户信息
            //todo
            //查询用户权限

            ResultData.success();
        } catch (Exception e) {
            resultData = ResultData.fail().message(e.getMessage());
        }
        return resultData;
    }

    /**
     * 解析字段
     * @param column
     * @param record
     * @param map
     * @return
     */
    private Map<String, Object> getColumnValue(Column column, Record record, Map<String, Object> map) {
        if (column.getTypeInfo().getOdpsType() == OdpsType.BIGINT) {
            String name = column.getName();
            Long colValue = record.getBigint(name);
            map.put(name, colValue);
        }
        if (column.getTypeInfo().getOdpsType() == OdpsType.STRING) {
            String name = column.getName();
            String colValue = record.getString(name);
            map.put(name, colValue);
        }
        if (column.getTypeInfo().getOdpsType() == OdpsType.DATETIME || column.getTypeInfo().getOdpsType() == OdpsType.DATE) {
            String name = column.getName();
            Date datetime = record.getDatetime(name);
            map.put(name, datetime);
        }
        return map;
     * 解析列和列值
     * @param record
     * @param schema
     * @return
     */
    private Map<String, Object> consumeRecord(Record record, TableSchema schema) {
        Map<String, Object> dataMap = new HashMap<>();
        for (int i = 0; i < schema.getColumns().size(); i++) {
            Column column = schema.getColumn(i);
            String name = column.getName();
            String colValue = null;
            switch (column.getTypeInfo().getOdpsType()) {
                case BIGINT:{
                    Long col_value = record.getBigint(i);
                    colValue = col_value == null ? null : col_value.toString();
                    break;
                }
                case STRING:{
                    String col_value = record.getString(i);
                    colValue = col_value == null ? null : col_value;
                    break;
                }
                case DATETIME:{
                    Date col_value = record.getDatetime(i);
                    colValue = col_value == null ? null : col_value.toString();
                    break;
                }
                default:
                    throw new RuntimeException("Unknown column type: " + column.getTypeInfo().getOdpsType());
            }
            dataMap.put(name, colValue);
        }
        return dataMap;
    }

    /**
     * 解析字段信息
     * @param column
     * @param record
     * @param dataMap
     * @return
     */
    private Map<String, Object> getColumnValue(Column column, Record record, Map<String, Object> dataMap) {
        String name = column.getName();
        Object colValue = null;
        if (column.getTypeInfo().getOdpsType() == OdpsType.BIGINT) {
            colValue = record.getBigint(name);
        }
        if (column.getTypeInfo().getOdpsType() == OdpsType.STRING) {
            colValue = record.getString(name);
        }
        if (column.getTypeInfo().getOdpsType() == OdpsType.DATETIME) {
            colValue = record.getDatetime(name);
        }
        dataMap.put(name, colValue);
        return dataMap;
    }

    private class DownloadThread implements Callable<Long> {
        private long id;
        private RecordReader recordReader;
        private TableSchema tableSchema;

        public DownloadThread(int id, RecordReader recordReader, TableSchema tableSchema) {
            this.id = id;
            this.recordReader = recordReader;
            this.tableSchema = tableSchema;
        }

        @Override
        public Long call() throws Exception {
            Long recordNum = 0L;
            try {
                Record record;
                while (null != (record = recordReader.read())) {
                    recordNum++;
                    System.out.println("Thread " + id + "\t");
                    consumeRecord(record, tableSchema);
                }
                recordReader.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
            return recordNum;
        }
    }
}
